package cn.hexcloud.dga.governance.assessor.quality;

import cn.hexcloud.dga.common.constant.MetaConst;
import cn.hexcloud.dga.governance.assessor.Assessor;
import cn.hexcloud.dga.governance.bean.AssessParam;
import cn.hexcloud.dga.governance.bean.GovernanceAssessDetail;
import cn.hexcloud.dga.governance.bean.GovernanceMetric;
import cn.hexcloud.dga.meta.bean.TableMetaInfo;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.commons.lang3.time.DateFormatUtils;
import org.apache.commons.lang3.time.DateUtils;
import org.apache.commons.math3.analysis.function.Abs;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import java.io.IOException;
import java.math.BigDecimal;
import java.net.URI;
import java.util.Date;

/**
 * @author jiangdan7
 */
@Component("PRODUCE_DATA_SIZE")
public class ProduceDataSizeAssessor extends Assessor {
    @Override
    public void checkProblem(GovernanceAssessDetail governanceAssessDetail, AssessParam assessParam) throws Exception {
        //提取参数
        GovernanceMetric governanceMetric = assessParam.getGovernanceMetric();
        JSONObject metricParamsJson = JSONObject.parseObject(governanceMetric.getMetricParamsJson());
        Integer days = metricParamsJson.getInteger("days");
        Integer upperLimit = metricParamsJson.getInteger("upper_limit");
        Integer lowerLimit = metricParamsJson.getInteger("lower_limit");

        TableMetaInfo tableMetaInfo = assessParam.getTableMetaInfo();
        String tableFsPath = tableMetaInfo.getTableFsPath();
        String tableFsOwner = tableMetaInfo.getTableFsOwner();

        String assessDate = assessParam.getAssessDate();

        // 按日分区的表
//        if (MetaConst.LIFECYCLE_TYPE_DAY.equals(tableMetaInfo.getTableMetaInfoExtra().getLifecycleType())) {
        // 获取分区字段信息，当前全为单分区字段dt=yyyy-MM-dd
        String partitionName = JSON.parseArray(tableMetaInfo.getPartitionColNameJson(), JSONObject.class).get(0).getString("name");
        Date partitionDate = DateUtils.addDays(DateUtils.parseDate(assessDate, "yyyy-MM-dd"), -1);
        String partitionDt = DateFormatUtils.format(partitionDate, "yyyy-MM-dd");
        // 获取昨天的数据量
        String path = tableFsPath + "/" + partitionName + "=" + partitionDt;
        Long curDateSize = getDataSizeByPath(path, tableFsOwner);
        // 获取近days天的平均数据量
        int partCountWithData = 0;
        Long partTotalDateSize = 0L;
        for (Integer i = 1; i <= days; i++) {
            Date otherPartitionDate = DateUtils.addDays(partitionDate, -i);
            String otherPartitionDt = DateFormatUtils.format(otherPartitionDate, "yyyy-MM-dd");
            String otherPath = tableFsPath + "/" + partitionName + "=" + otherPartitionDt;
            Long otherDateSize = getDataSizeByPath(otherPath, tableFsOwner);
            if (otherDateSize > 0) {
                partCountWithData++;
                partTotalDateSize += otherDateSize;
            }
        }
        // 未有历史数据，评估退出
        if (partCountWithData == 0) {
            return;
        }
        long race = curDateSize * partCountWithData * 100 / partTotalDateSize;
        if (race > 100 + upperLimit) {
            governanceAssessDetail.setAssessScore(BigDecimal.ZERO);
            governanceAssessDetail.setAssessProblem(String.format("昨天的新增数据量超过了近%d天的平均新增数据量的阈值%d%%，超出%d%%", days, upperLimit, race - 100));
        } else if (race < 100 - lowerLimit) {
            governanceAssessDetail.setAssessScore(BigDecimal.ZERO);
            governanceAssessDetail.setAssessProblem(String.format("昨天的新增数据量低出了近%d天的平均新增数据量的阈值%d%%，低出%d%%", days, lowerLimit, 100 - race));
        }
    }
//    }

    private Long getDataSizeByPath(String path, String tableFsOwner) throws Exception {
        FileSystem fileSystem = FileSystem.get(new URI(path), new Configuration(), tableFsOwner);
        if (fileSystem.exists(new Path(path))) {
            FileStatus[] fileStatuses = fileSystem.listStatus(new Path(path));
            return getFileDataSize(fileSystem, 0L, fileStatuses);
        }
        return 0L;
    }

    private Long getFileDataSize(FileSystem fileSystem, long curSize, FileStatus[] fileStatuses) throws IOException {
        for (FileStatus fileStatus : fileStatuses) {
            if (fileStatus.isDirectory()) {
                FileStatus[] subFileStatuses = fileSystem.listStatus(fileStatus.getPath());
                curSize = getFileDataSize(fileSystem, curSize, subFileStatuses);
            } else {
                curSize += fileStatus.getLen();
            }
        }
        return curSize;
    }
}
